package com.czk.zookeeper.zookeepercli.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * zookeeper实现分布式锁
 * 由于zookeeper是创建节点是线程安全（同步锁）的，所以其能通过临时（同一个连接能看到）有序节点来实现分布式锁
 * 由于写操作都会在master节点上进行，所以集群也不会有线程安全问题
 *
 * @Author:ChenZhangKun
 * @Date: 2021/1/5 21:47
 */
public class MyLock {
    String IP = "192.168.25.133:2181";
    CountDownLatch countDownLatch = new CountDownLatch(1);
    ZooKeeper zooKeeper;
    /**
     * 根节点
     */
    private static final String LOCK_ROOT_PATH = "/locks";
    private static final String LOCK_NODE_NAME = "Lock_";
    private String lockPath;

    /**
     * 构造初始化连接
     *
     * @throws InterruptedException
     * @throws IOException
     */
    public MyLock() throws InterruptedException, IOException {
        zooKeeper = new ZooKeeper(IP, 50000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功");
                    // 计数器减一
                    countDownLatch.countDown();
                }
            }
        });
        countDownLatch.await();
    }

    /**
     * 尝试获取锁
     *
     * @throws Exception
     */
    public void acquireLock() throws Exception {
        createLock();
        attemptLock();
    }

    /**
     * 创建锁节点
     */
    public void createLock() throws Exception {
        // 判断Locks是否存在，不存在则创建
        Stat stat = zooKeeper.exists(LOCK_ROOT_PATH, false);
        // 节点等于null
        if (stat == null) {
            // 创建主节点
            zooKeeper.create(LOCK_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        // 创建临时有序节点
        lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("节点路径" + lockPath);
    }

    /**
     * 尝试获取锁
     */
    public void attemptLock() throws Exception {
        // 获取所有子节点
        List<String> children = zooKeeper.getChildren(LOCK_ROOT_PATH, false);
        // 堆子节点进行排序
        Collections.sort(children);
        // 临时有序节点的索引
        int i = children.indexOf(lockPath.substring(LOCK_NODE_NAME.length() + 1));
        // 临时节点在第一位，获取锁
        if (i == 0) {
            System.out.println("获取锁成功");
            return;
        } else if (i > 0) {
            // 获取上一个节点的节点路径
            String prePath = children.get(i - 1);
            // 监视上一个节点
            Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + prePath, watcher);
            if (stat == null) {
                // 尝试获取锁
                attemptLock();
            } else {
                // 上一个节点在等待或执行
                // 处于等待状态
                synchronized (watcher) {
                    watcher.wait();
                }
                // 线程苏醒，接着获取锁
                attemptLock();
            }
        }
    }

    public void releaseLock() throws KeeperException, InterruptedException {
        // 释放锁
        zooKeeper.delete(lockPath, -1);
        // 关闭zookeeper
        zooKeeper.close();
        //
        System.out.println("锁已经释放");
    }

    /**
     * 监视器对象
     */
    Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
                // 唤醒锁
                synchronized (watcher) {
                    watcher.notifyAll();
                }
            }
        }
    };

    public static void main(String[] args) throws Exception {
        MyLock lock = new MyLock();
        lock.createLock();
    }
}
